package net.bwie.jtp.dim.trade.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.configuration.Configuration;

import java.util.HashSet;
import java.util.Set;

public class DimDataFilterFunction extends RichFilterFunction <String> {

    private final Set<String> dimSet;
    private final Set<String> operateTypeSet=new HashSet<String>(){{
         add("insert");add("update");add("delete");
    }};

    public DimDataFilterFunction(Set<String> dimSet) {
        this.dimSet = dimSet;
    }
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public boolean filter(String value) throws Exception {
        JSONObject jsonObject = JSON.parseObject(value);
        //获取表名称
        String tableName = jsonObject.getString("table_name");
//        System.out.println("dimSet"+dimSet);
//        System.out.println("tableName"+tableName);
        //操作类型operate_type
        String operateType = jsonObject.getString("operate_type");
        if (tableName.equals("user_info")){
            System.out.println(tableName+operateType);

        }
        return dimSet.contains(tableName)&&operateTypeSet.contains(operateType);
    }
}
